//////////////////////////////////////////////////
// EASIM project
// easim.net.ACK
// 6/20/2007 9:25 AM
// Cong Liu
// gzcong@gmail.com

// Copyright (C) 2007
// All rights reserved.
// Refer to LICENSE of JIST (Java In Simulation Time) for terms and conditions of use.

package easim.omar;

import easim.net.DtnNodeInterface;
import easim.net.Clock;
import easim.net.Timer;

import jist.swans.Constants;
import jist.swans.misc.Message;
import jist.runtime.JistAPI.DoNotRewrite;

import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Iterator;

public class ACK implements DtnNodeInterface, DoNotRewrite
{

  public static long INIT_TIME = - 1000 * Constants.SECOND;
  public static long CLEANER_INTERVAL = 10 * Constants.SECOND;
  public static long ACK_INTERVAL = 5 * Constants.SECOND;
  public static double ACK_PROPA_PROB = 0.5;
  public static long MAX_ACK_TTL = 20 * Constants.SECOND;
  public static int MAX_ACK_NUM = 64;
    
  /** class TableItem */
  public static class AckItem implements DoNotRewrite
  {
    public int srcId = -1;
    public int msgId = -1;
    public long addTime = INIT_TIME;
    public long timeout = INIT_TIME;

    public String toString() {
      return srcId + "-" + msgId + "(" + timeout + ")";
    }
    public boolean equals(Object obj) {
      if (! (obj instanceof AckItem)) return false;
      AckItem ack = (AckItem)obj;
      return (srcId == ack.srcId && msgId == ack.msgId);
    }
    public int hashCode() {
      return (msgId * 1000 + srcId);
    }
    public void copyTo(AckItem ack) {
      ack.srcId = srcId;
      ack.msgId = msgId;
      ack.addTime = addTime;
      ack.timeout = timeout;
    }
    public AckItem getClone() {
      AckItem ack = new AckItem();
      copyTo(ack);
      return ack;
    }
  } // end class TableItem

  /** class NodeInfo */
  protected static class NodeInfo implements DoNotRewrite
  {
    public int id = -1;
    public long lastUpdateSentTime = INIT_TIME;
    public long lastUpdateRecvTime = INIT_TIME;
  } // end class NodeInfo

  /** class ContactInfo */
  protected static class ContactInfo implements DoNotRewrite
  {
    public int id = -1;
    public long encounterTime = INIT_TIME;
  } // end class ContactInfo

  /** class UpdateMessage */
  public static class UpdateMessage implements Message, DoNotRewrite
  {
    public long time;
    public AckItem[] acks;
    public int getSize() {
      return 8 + 16 * acks.length;
    }
    public void getBytes(byte[] msg, int offset) {
      throw new RuntimeException("no implementation.");
    }
    // for rollback, not part of the message exactly.
    public HashMap rbInfo;
  } // end class UpdateMessage

  /** class Cleaner */
  public class Cleaner implements Timer, DoNotRewrite
  {
    public void start() {
      Clock.start(this, CLEANER_INTERVAL, false);
    }
    public void timeout() {
      long curTime = Clock.getTime();
      Iterator it = acks.iterator();
      while (it.hasNext()) {
        AckItem ack = (AckItem)it.next();
        if (ack.timeout <= curTime) it.remove();
      }
      start();
    }
  } // end class Cleaner

  ///////////////////////////
  // variables
  //

  protected Node node;
  protected HashSet acks = new HashSet();
  protected HashMap nodes = new HashMap();
  protected HashMap contacts = new HashMap();
  protected long updateTime = INIT_TIME;
  protected Cleaner cleaner = new Cleaner();

  public ACK(Node node) {
    this.node = node;
    cleaner.start();
  }

  public void onContactDetected(int contactId) {
    node.startJitterTimer();
    ContactInfo contact = new ContactInfo();
    contact.id = contactId;
    contact.encounterTime = Clock.getTime();
    contacts.put(new Integer(contact.id), contact);
    if (! nodes.containsKey(new Integer(contactId))) {
      NodeInfo info = new NodeInfo();
      info.id = contactId;
      nodes.put(new Integer(info.id), info);
    }
  }

  public void onContactBroken(int contactId) {
    contacts.remove(new Integer(contactId));
  }

  public void receiveMessage(Message message, int srcId) {
    UpdateMessage uMsg = (UpdateMessage)message;
    long curTime = Clock.getTime();
    NodeInfo info = (NodeInfo)nodes.get(new Integer(srcId));
    info.lastUpdateRecvTime = curTime;
    //info.upToDate = uMsg.upToDate;
    incrementalUpdate(uMsg.acks, uMsg.time);
  }

  public Message nextMessageToSend() {
    long curTime = Clock.getTime();
    UpdateMessage uMsg = new UpdateMessage();
    uMsg.time = curTime;
    long lastUpdateTime = curTime;
    Iterator it = contacts.values().iterator();
    boolean needToSend = false;
    while (it.hasNext()) {
      ContactInfo contact = (ContactInfo)it.next();
      NodeInfo info = (NodeInfo)nodes.get(new Integer(contact.id));
      if (lastUpdateTime > info.lastUpdateSentTime + ACK_INTERVAL) {
        lastUpdateTime = info.lastUpdateSentTime;
      }
      if (contact.encounterTime > info.lastUpdateSentTime + ACK_INTERVAL) {
        needToSend = true;
      }
    }
    if (lastUpdateTime > updateTime && (! needToSend)) {
      return null;
    }
    Iterator all = acks.iterator();
    LinkedList toSend = new LinkedList();
    long latestUpdateTime = lastUpdateTime;
    while (all.hasNext()) {
      AckItem ack = (AckItem)all.next();
      if (ack.timeout < curTime) {
        all.remove();
      }
      else if (ack.addTime >= lastUpdateTime) {
        AckItem ack2 = ack.getClone();
        if (ack2.timeout > curTime + MAX_ACK_TTL) {
          ack2.timeout = curTime + MAX_ACK_TTL;
        }
        toSend.addLast(ack2);
        if (latestUpdateTime < ack.addTime) {
          latestUpdateTime = ack.addTime;
        }
      }
    }
    if (toSend.size() == 0 && (! needToSend)) {
      return null;
    }
    uMsg.acks = (AckItem[])toSend.toArray(new AckItem[0]);
    if (uMsg.acks.length > MAX_ACK_NUM) {
      for (int i = 0; i < uMsg.acks.length - 1; ++ i) {
        boolean swapped = false;
        for (int j = uMsg.acks.length - 1; j > i; -- j) {
          if (uMsg.acks[j - 1].timeout > uMsg.acks[j].timeout) {
            AckItem tmp = uMsg.acks[j - 1];
            uMsg.acks[j - 1] = uMsg.acks[j];
            uMsg.acks[j] = tmp;
            swapped = true;
          }
        }
        if (! swapped) break;
      }
      AckItem[] acks2 = uMsg.acks;
      uMsg.acks = new AckItem[MAX_ACK_NUM];
      for (int i = 0; i < MAX_ACK_NUM; ++ i) {
        uMsg.acks[i] = acks2[acks2.length - MAX_ACK_NUM + i];
      }
    }
    uMsg.rbInfo = new HashMap();
    it = contacts.values().iterator();
    while (it.hasNext()) {
      ContactInfo contact = (ContactInfo)it.next();
      NodeInfo info = (NodeInfo)nodes.get(new Integer(contact.id));
      if (info.lastUpdateSentTime <= latestUpdateTime ||
          info.lastUpdateSentTime < contact.encounterTime) {
        uMsg.rbInfo.put(new Integer(info.id), new Long(info.lastUpdateSentTime));
        info.lastUpdateSentTime = curTime;
      }
    }
    ++ Stats.send.ackMsgs;
    Stats.send.ackMsgSize += uMsg.getSize();
    //System.out.println(toString(node.getId(), uMsg.acks));
    return uMsg;
  }

  public void onAckMessage(Message message) {
    UpdateMessage uMsg = (UpdateMessage)message;
    ++ Stats.recv.ackMsgs;
    Stats.recv.ackMsgSize += uMsg.getSize();
  }

  /** rollback the update information */
  public void onDropMessage(Message message, int[] dstIds) {
    UpdateMessage uMsg = (UpdateMessage)message;
    boolean dropped = false;
    for (int i = 0; i < dstIds.length; ++ i) {
      int unacked = dstIds[i];
      Long lastUpdateSentTime = (Long)uMsg.rbInfo.get(new Integer(unacked));
      if (lastUpdateSentTime == null) continue;
      NodeInfo info = (NodeInfo)nodes.get(new Integer(unacked));
      info.lastUpdateSentTime = lastUpdateSentTime.longValue();
      dropped = true;
    }
    if (! dropped) {
      onAckMessage(message);
    }
  }

  /** update the mirror table of the neighbor */
  public void incrementalUpdate(AckItem[] updates, long srcTime) {
    long curTime = Clock.getTime();
    for (int i = 0; i < updates.length; ++ i) {
      AckItem ack = updates[i];
      if (! acks.contains(ack)) {
        ack = ack.getClone();
        ack.addTime = curTime;
        ack.timeout = ack.timeout + (curTime - srcTime);
        if (node.data.removeData(ack.srcId, ack.msgId)) {
          acks.add(ack);
          updateTime = curTime;
        }
        else {
          double p = Constants.random.nextDouble();
          if (p < ACK_PROPA_PROB) {
            acks.add(ack);
            updateTime = curTime;
          }
        }
      }
    }
  }

  public boolean containsAck(int srcId, int msgId) {
    AckItem ack = new AckItem();
    ack.srcId = srcId;
    ack.msgId = msgId;
    return acks.contains(ack);
  }

  public boolean addAck(int srcId, int msgId, long timeout) {
    AckItem ack = new AckItem();
    ack.srcId = srcId;
    ack.msgId = msgId;
    if (acks.contains(ack)) return false;
    ack.timeout = timeout;
    ack.addTime = Clock.getTime();
    updateTime = ack.addTime;
    acks.add(ack);
    node.startJitterTimer();
    return true;
  }

  public boolean updateReceived(int contactId) {
    NodeInfo info = (NodeInfo)nodes.get(new Integer(contactId));
    if (info == null) return false;
    ContactInfo contact = (ContactInfo)contacts.get(new Integer(contactId));
    if (contact == null) return false;
    return (info.lastUpdateRecvTime > contact.encounterTime);
  }

  public String toString() {
    String buf = "ACK=\r\n";
    Iterator it = acks.iterator();
    while (it.hasNext()) {
      AckItem ack = (AckItem)it.next();
      buf += ack + "\r\n";
    }
    return buf;
  }

  public static String toString(int id, AckItem[] acks) {
    String buf = "[" + id + "] ACK=\r\n";
    for (int i = 0; i < acks.length; ++ i) {
      buf += acks[i] + "\r\n";
    }
    return buf;
  }

}
